package com.tc.vms.consumer;

import com.google.protobuf.InvalidProtocolBufferException;
import com.rabbitmq.client.*;
import com.tc.vms.MessageWrap;
import com.tc.vms.VMSEventArgs;
import com.tc.vms.encoding.Message;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.iface.IContext;
import com.tc.vms.iface.ISubscriber;
import com.tc.vms.message.MessageConst;
import com.tc.vms.producer.VmsProducerRmqMultiQueueImpl;
import com.tc.vms.utils.ServiceThread;
import com.tc.vms.utils.Utils;
import com.tc.vms.utils.VmsThreadFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

import static com.tc.vms.utils.Utils.MAX_DELAY_LEVEL;
import static com.tc.vms.utils.Utils.MIN_DELAY_LEVEL;

/**
 * Created by yonghua.zhang on 2015/12/14.
 */
public class VmsConsumerRmqImpl extends AbstractConsumer implements ISubscriber {

    protected final ConnectionFactory factory;
    protected Connection connection;
    protected Channel channel;
    protected Address[] destAddresses;

    protected ExecutorService consumeService;
    protected final BlockingQueue<Runnable> consumeQueue;
    protected Map<String, Integer> physicalQueueMap;
    protected boolean isQueueSharding;
    protected ServiceThread pullMessageService;
    protected final ServiceThread procShutdownSignal;
    protected volatile boolean hasShutdownSignal = false;
    protected QueueingConsumer queueingConsumer;
    protected volatile boolean isConsumerRunning = false;
    private final String retryQueueName;
    private final String dieQueueName;
    private final Map<Integer, String> retryExchangeNameCache;

    public VmsConsumerRmqImpl(VmsConsumerProxy vmsConsumerProxy) {
        super(vmsConsumerProxy);
        retryQueueName = queueMetaData.getRetryQueueName();
        dieQueueName = queueMetaData.getDLQName();
        factory = new ConnectionFactory();
        consumeQueue = new LinkedBlockingQueue<Runnable>(4096);
        retryExchangeNameCache = new HashMap<Integer, String>();
        procShutdownSignal = new ServiceThread() {
            @Override
            public String getServiceName() {
                return "Process-consumer-shutdown-signal-thread_" + System.currentTimeMillis();
            }

            @Override
            public void run() {
                while (true) {
                    try {
                        if (hasShutdownSignal) {
                            TimeUnit.MILLISECONDS.sleep(3000);
                            LOGGER.error(
                                    "[VmsConsumerRmqImpl] Rabbitmq channel exception, and recovery channel. curThreadId = [{}]",
                                    Thread.currentThread().getName());
                            synchronized (VmsConsumerRmqImpl.this) {
                                if (isOpen) {
                                    while (channel == null || !channel.isOpen()) {
                                        try {
                                            innerShutdown();
                                            LOGGER.info(
                                                    "[VmsConsumerRmqImpl] RestoreShutdown - innerShutdown operation successfully.");
                                            createChannel();
                                            LOGGER.info(
                                                    "[VmsConsumerRmqImpl] RestoreShutdown - createChannel operation successfully.");
                                            break;
                                        } catch (Exception e) {
                                            TimeUnit.MILLISECONDS.sleep(3000);
                                            LOGGER.error(
                                                    "[VmsConsumerRmqImpl] Rabbitmq channel exception, Recovery channel failed and retry after 3s interval.", e);
                                        }
                                    }

                                    while (channel != null && channel.isOpen()) {
                                        try {
                                            stopConsume();
                                            TimeUnit.MILLISECONDS.sleep(3000);
                                            startConsume();
                                            LOGGER.info(
                                                    "[VmsConsumerRmqImpl] RestoreShutdown - startConsume operation successfully.");
                                            hasShutdownSignal = false;
                                            break;
                                        } catch (Exception e) {
                                            TimeUnit.MILLISECONDS.sleep(3000);
                                            LOGGER.error(
                                                    "[VmsConsumerRmqImpl] Rabbitmq consume exception, Recovery consume failed and retry after 3s interval.", e);
                                        }
                                    }
                                }
                            }
                        } else {
                            TimeUnit.MILLISECONDS.sleep(3000);
                        }
                    } catch (InterruptedException e) {
                        LOGGER.warn("Process-consumer-shutdown-signal-thread shutdown.");
                        break;
                    }
                }
            }
        };
    }

    @Override
    public synchronized boolean init(IContext context) throws VmsClientException {
        try {
            destAddresses = queueMetaData.getBrokerAddress();
            factory.setVirtualHost(queueMetaData.getVhost());
            factory.setUsername(queueMetaData.getUserName());
            factory.setPassword(queueMetaData.getUserPasswd());
            //factory.setAutomaticRecoveryEnabled(true);
            factory.setRequestedHeartbeat(Utils.RMQ_HEART_BEAT_INTERVAL);

            String logicQueueName = queueMetaData.getRealQueueNmOnBroker();
            Set<String> tmpShardingTagSet = queueMetaData.getQueueShards();
            HashMap<String, Integer> tmpQueueMap = new HashMap<String, Integer>(tmpShardingTagSet.size());
            if (tmpShardingTagSet.isEmpty()) {
                isQueueSharding = false;
                tmpQueueMap.put(logicQueueName, null);
            } else {
                for (String shardingTag : tmpShardingTagSet) {
                    tmpQueueMap.put(Utils.joinShardsTags(logicQueueName, shardingTag),
                            Utils.getBrokerIdByShardingTag(shardingTag));
                }
                isQueueSharding = true;
            }
            physicalQueueMap = Collections.unmodifiableMap(tmpQueueMap);
            return true;
        } catch (Exception e) {
            throw new VmsClientException("VmsConsumerRmqImpl init failed.", e);

        }
    }

    @Override
    public synchronized boolean start() throws VmsClientException {
        if (!isOpen) {
            try {
                createChannel();
                declareDelayQueues();
                startConsume();
                procShutdownSignal.start();
                isOpen = true;
            } catch (Exception ex) {
                innerShutdown();
                procShutdownSignal.shutdown(true);
                throw new VmsClientException("VmsConsumerRmqImpl start failed.", ex);
            }

        }
        return isOpen;
    }

    protected void innerShutdown() {
        try {
            stopConsume();
        } catch (Exception ex) {
        }
        if (channel != null) {
            try {
                channel.close();
            } catch (Exception e) {
            }
            channel = null;
        }
        if (connection != null) {
            try {
                connection.close();
            } catch (Exception e) {
            }
            connection = null;
        }
    }

    @Override
    public synchronized void shutdown() {
        if (isOpen) {
            isOpen = false;
            hasShutdownSignal = false;
            procShutdownSignal.shutdown(true);
            innerShutdown();
        }
    }

    private List<Map.Entry<String, Integer>> shufflePhysicalQueues() {
        ArrayList<Map.Entry<String, Integer>> entries = new ArrayList<Map.Entry<String, Integer>>();
        for (Map.Entry entry : physicalQueueMap.entrySet()) {
            entries.add(entry);
        }
        Collections.shuffle(entries);
        return entries;
    }

    protected void stopConsume() {
        isConsumerRunning = false;
        if (pullMessageService != null) {
            pullMessageService.shutdown(true);
            pullMessageService = null;
        }

        if (consumeService != null) {
            consumeService.shutdown();
            try {
                consumeService.awaitTermination(5, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
            }
            consumeService = null;
        }
        queueingConsumer = null;
        consumeQueue.clear();
    }

    protected void commit(ConsumeStatus cs, MessageWrap message, VmsRmqConsumerContext vmsRmqConsumerContext) {
        if (isConsumerRunning && channel != null && !subQoS.autoCommit()) {
            final long ackIdx = vmsRmqConsumerContext.getAckIdx();
            synchronized (channel) {
                try {
                    switch (cs) {
                        case CONSUME_SUCCESS:
                            channel.basicAck(ackIdx, false);
                            break;
                        case RECONSUME_LATER:
                            if (message.getReconsumeTimes() <= 0) {
                                MessageConst.setMessageReconsumeTimes(message, 0);
                                message.setDelayLevel(MIN_DELAY_LEVEL);
                            }

                            int delayLevel = vmsRmqConsumerContext.getDelayLevelWhenReconsume() > MAX_DELAY_LEVEL ?
                                    MAX_DELAY_LEVEL :
                                    vmsRmqConsumerContext.getDelayLevelWhenReconsume();
                            delayLevel = delayLevel >= MIN_DELAY_LEVEL ? delayLevel : MIN_DELAY_LEVEL;

                            boolean isRetryMaxTimes = message.getReconsumeTimes() >= MAX_DELAY_LEVEL;
                            int consumeTimes = isRetryMaxTimes ? MAX_DELAY_LEVEL : message.getReconsumeTimes();
                            message.setDelayLevel(consumeTimes + 1);
                            MessageConst.setMessageReconsumeTimes(message, consumeTimes + 1);
                            Message.VMSMessage vmsMessage = VmsProducerRmqMultiQueueImpl.buildVmsMessage(message);
                            try {
                                if (isRetryMaxTimes) {
                                    channel.queueDeclare(dieQueueName, true, false, false, null);
                                    channel.basicPublish("", dieQueueName, MessageProperties.PERSISTENT_TEXT_PLAIN,
                                            vmsMessage.toByteArray());
                                } else {
                                    String tmpDelayExchangeName = retryExchangeNameCache.get(delayLevel);
                                    channel.basicPublish(tmpDelayExchangeName, queueMetaData.getRealQueueNmOnBroker(),
                                            MessageProperties.PERSISTENT_TEXT_PLAIN, vmsMessage.toByteArray());
                                }
                                channel.waitForConfirms(5000);
                                channel.basicAck(ackIdx, false);
                            } catch (Exception ex) {
                                LOGGER.warn("Sendback message exception and prefer to nack the message.", ex);
                                channel.basicNack(ackIdx, false, true);
                            }
                            break;
                        case CONSUME_DISCARD:
                        default:
                            channel.basicNack(ackIdx, false, false);
                            break;
                    }
                } catch (Exception ex) {
                    LOGGER.warn("COMMIT - Consumer ack exception. AckIdx = {}, channel is Active = {}", ackIdx,
                            channel.isOpen(), ex);
                    if (channel.isOpen()) {
                        try {
                            channel.basicRecover(true);
                        } catch (IOException e) {
                            LOGGER.error("COMMIT - Consumer basicRecover failed.", e);
                        }
                    }
                }
            }
        }
    }

    protected void startConsume() throws VmsClientException {
        isConsumerRunning = true;
        try {
            queueingConsumer = new QueueingConsumer(channel);

            int iConsumer = 0;
            for (String queueName : physicalQueueMap.keySet()) {
                channel.basicConsume(queueName, subQoS.autoCommit(), instName + (iConsumer++), queueingConsumer);
            }
            //consuming retry queue.
            channel.basicConsume(retryQueueName, subQoS.autoCommit(), instName + (iConsumer++), queueingConsumer);

            consumeService = new ThreadPoolExecutor(//
                    this.consumingThreadNum,//
                    this.consumingThreadNum * 2,//
                    1000 * 60,//
                    TimeUnit.MILLISECONDS,//
                    consumeQueue,//
                    new VmsThreadFactory("Vms_Rabbitmq_Consume_Thread"));

            pullMessageService = new ServiceThread() {
                @Override
                public void run() {
                    while (isConsumerRunning) {
                        try {
                            final QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
                            while (vmsConsumerProxy.isSuspend()) {
                                LOGGER.debug("Consuming service is suspend by vmscenter");
                                Utils.sleep5s();
                            }
                            consumeService.submit(new Runnable() {
                                @Override
                                public void run() {
                                    ConsumeStatus cs = ConsumeStatus.CONSUME_DISCARD;
                                    VMSEventArgs consumeMessage = null;
                                    MessageWrap message = null;
                                    final VmsRmqConsumerContext vmsRmqConsumerContext = new VmsRmqConsumerContext(0,
                                            delivery.getEnvelope().getDeliveryTag());
                                    try {

                                        try {
                                            message = parseFromPayload(delivery);
                                            consumeMessage = new VMSEventArgs(true, message);
                                        } catch (Exception ex) {
                                            message = new MessageWrap(delivery.getBody());
                                            consumeMessage = new VMSEventArgs(false, message);
                                        }

                                        try {
                                            vmsRmqConsumerContext.setDelayLevelWhenReconsume(message.getDelayLevel());
                                            cs = callback.consumeMessage(vmsRmqConsumerContext, consumeMessage);
                                        } catch (Exception ex) {
                                            LOGGER.warn("invoke callback function failed.", ex);
                                        }
                                    } finally {
                                        if (isConsumerRunning) {
                                            commit(cs, message, vmsRmqConsumerContext);
                                        }
                                    }
                                }
                            });

                        } catch (InterruptedException e) {
                            LOGGER.warn("{} is interrupted.", getServiceName());
                            break;
                        } catch (ConsumerCancelledException cce) {
                            LOGGER.warn("pullMessageService rabbitmq ConsumerCancelledException, restart consuming. ->", cce);
                            hasShutdownSignal = true;
                            break;
                        } catch (ShutdownSignalException sse) {
                            LOGGER.warn("pullMessageService rabbitmq channel be closed,  restart consuming. -> ", sse);
                            hasShutdownSignal = true;
                            break;
                        } catch (Exception otherEx) { //
                            LOGGER.warn("pullMessageService exception. -> ", otherEx);
                            Utils.sleep3s();
                        }
                    }
                }

                @Override
                public String getServiceName() {
                    return "Rabbitmq_PullMessage_Service_" + String.valueOf(System.currentTimeMillis());
                }
            };
            pullMessageService.start();
            LOGGER.debug("pullMessageService has been started. threadId=[{}]", Thread.currentThread().getId());
        } catch (Exception ex) {
            stopConsume();
            throw new VmsClientException("VmsConsumerRmqImpl startConsume failed.", ex);
        }
    }

    protected void createChannel() throws VmsClientException {
        if (channel == null) {
            try {
                if (connection == null || !connection.isOpen()) {
                    if (!isQueueSharding) {
                        LOGGER.debug(
                                "VmsConsumerRmqImpl createChannel for no-sharding queue. channelName=[{}], address=[{}]",
                                queueMetaData.getRealQueueNmOnBroker(), queueMetaData.getBrokerCluster());
                        Address[] addresses = Utils.shuffle(queueMetaData.getBrokerAddress());
                        connection = factory.newConnection(addresses);
                    } else {
                        LOGGER.debug(
                                "VmsConsumerRmqImpl createChannel for sharding queue. channelName=[{}], address=[{}]",
                                queueMetaData.getRealQueueNmOnBroker(), queueMetaData.getBrokerCluster());
                        List<Map.Entry<String, Integer>> physicalQueues = shufflePhysicalQueues();
                        Exception lastException = null;
                        boolean createdOk = false;
                        for (Map.Entry<String, Integer> shardAddr : physicalQueues) {
                            try {
                                int brokerIdx = shardAddr.getValue() % destAddresses.length;
                                Address[] tmpHostAddrs = new Address[]{destAddresses[brokerIdx]};
                                connection = factory.newConnection(tmpHostAddrs);
                                createdOk = true;
                                break;
                            } catch (Exception e) {
                                lastException = e;
                            }
                        }
                        if (!createdOk) {
                            throw (lastException != null) ?
                                    new VmsClientException("failed to connect." + lastException, lastException) :
                                    new VmsClientException("failed to connect.");
                        }

                    }
                }

                channel = connection.createChannel();
                channel.confirmSelect();
                channel.addShutdownListener(new ShutdownListener() {
                    @Override
                    public synchronized void shutdownCompleted(ShutdownSignalException cause) {
                        if (!hasShutdownSignal) {
                            hasShutdownSignal = true;
                            LOGGER.warn("Rabbitmq channel has been shutdown by accident.");
                        }
                    }
                });

                if (subQoS.prefetchCount() > 0) {
                    channel.basicQos(subQoS.prefetchCount());
                } else {
                    channel.basicQos(64);
                }
            } catch (Exception e) {
                innerShutdown();
                LOGGER.error("Create Rabbitmq channel failed.", e);
                throw new VmsClientException("Create Rabbitmq channel failed.", e);
            }
        } else if (!channel.isOpen()) {
            innerShutdown();
            throw new VmsClientException("Create Rabbitmq channel failed.");
        }
    }

    public final static MessageWrap parseFromPayload(final QueueingConsumer.Delivery delivery)
            throws InvalidProtocolBufferException {
        MessageWrap message = null;
        try {
            Message.VMSMessage pbMessage = Message.VMSMessage.parseFrom(delivery.getBody());
            message = new MessageWrap(pbMessage.getPayload().toByteArray());
            message.setMessageId(pbMessage.getMessageId());

            if (pbMessage.getPropertiesList() != null) {
                for (Message.KeyPair kp : pbMessage.getPropertiesList()) {
                    message.setAttribute(kp.getKey(), kp.getValue());
                }
            }
            if (pbMessage.getRoutingKeysCount() > 0) {
                for (String key : pbMessage.getRoutingKeysList()) {
                    message.addRoutingKey(key);
                }
            }
            return message;
        } catch (InvalidProtocolBufferException e) {
            LOGGER.warn("Fail to parse protobuf message and set message = null, Exception : {}", e, e);
            throw e;
        }
    }

    //create delay/retry/dlq queues
    protected void declareDelayQueues() throws VmsClientException {
        if (channel != null && channel.isOpen()) {
            try {
                //if queue not exists, throw exception, don't create retry queue
                String anyQueue = physicalQueueMap.keySet().toArray(new String[0])[0];
                channel.queueDeclarePassive(anyQueue);

                parseDelayLevel();
                Map<String, Object> args = new HashMap<String, Object>(2);
                channel.queueDeclare(retryQueueName, true, false, false, null);

                for (Integer iLvl : this.delayLevelTable.keySet()) {
                    Long delayTime = this.delayLevelTable.get(iLvl);
                    String tmpDelayExchangeName = iLvl + ".vms.delay.exchange";
                    String tmpRedeliveryExchangeName = iLvl + ".vms.redelivery.exchange";
                    String tmpDelayQueueName = iLvl + ".vms.delay.queue";
                    channel.exchangeDeclare(tmpDelayExchangeName, "fanout", true, false, false, null);
                    channel.exchangeDeclare(tmpRedeliveryExchangeName, "direct", true, false, false, null);
                    retryExchangeNameCache.put(iLvl, tmpDelayExchangeName);

                    args.put("x-message-ttl", delayTime);
                    args.put("x-dead-letter-exchange", tmpRedeliveryExchangeName);
                    channel.queueDeclare(tmpDelayQueueName, true, false, false, args);
                    channel.queueBind(tmpDelayQueueName, tmpDelayExchangeName, "");
                    channel.queueBind(retryQueueName, tmpRedeliveryExchangeName, //死信队列 过期 发往retryQueueName
                            queueMetaData.getRealQueueNmOnBroker());
                }
            } catch (Exception e) {
                throw new VmsClientException("declareDelayQueues failed.", e);
            }
        }
    }

}
